<?php
defined('BASEPATH') or exit('No direct script access allowed');

class Rabbitmq
{

    /**    
     * @var \AMQPConnection
     */
    protected $_connection;

    /**    
     * @var \AMQPChannel
     */
    protected $_channel;
    protected $connect_conf;
    protected $is_delay_message = false;

    const AMQP_EX_DELAY_MESSAGE = 'x-delayed-message';

    public function __construct($params = [])
    {
        $config_name = $params['conf'] ?? 'default';
        $CI = &get_instance();
        $CI->config->load('rabbit_mq');
        $this->connect_conf = $CI->config->item($config_name);
        // writeln(json_encode($this->connect_conf));
    }

    public function get_connection($conf = null)
    {
        $_conf = $conf ?? $this->connect_conf;
        if ($this->_connection instanceof AMQPConnection) return $this->_connection;
        $this->_connection = new AMQPConnection($_conf);
        $this->_connection->connect();
        return $this->_connection;
    }

    /**
     *
     * @author yangmingzhi
     * @date 2021-08-11
     * @return \AMQPExchange
     */
    public function get_exchage($exchange_name = '', $type = AMQP_EX_TYPE_DIRECT)
    {

        $exchange = new AMQPExchange($this->get_channel());
        $exchange->setName($exchange_name);
        $exchange->setFlags(AMQP_DURABLE);        

        if ($this->is_delay_message) {
            //延时插件命名方式
            $exchange->setType(self::AMQP_EX_DELAY_MESSAGE);
            $args = [
                'x-delayed-type' => $type,
            ];
            $exchange->setArguments($args);
        } else {
            $exchange->setType($type);
        }


        $exchange->declareExchange();

        return $this->_exhange = $exchange;
    }

    /**
     *
     * @author yangmingzhi
     * @date 2021-08-11
     * @return \AMQPChannel
     */
    public function get_channel()
    {
        if ($this->_channel instanceof AMQPChannel) return $this->_channel;
        $this->_channel = new AMQPChannel($this->get_connection());
        $this->_channel->setPrefetchCount(1);
        // $this->_channel->qos(0, 1);
        return $this->_channel;
    }


    /**
     *
     * @author yangmingzhi
     * @date 2021-08-11
     * @return \AMQPQueue
     */
    public function get_queue($queue_name, $exchange_name, $routing_key)
    {
        $queue = new \AMQPQueue($this->get_channel());
        $queue->setName($queue_name);
        $queue->setFlags(AMQP_DURABLE);
        $args = [
            // 'x-message-ttl' => 6000,
            'x-dead-letter-exchange' => 'DL_EX',
            'x-dead-letter-routing-key' => 'DL_RK',
        ];
        $queue->setArguments($args);
        $queue->declareQueue();
        if (is_array($routing_key)) {
            foreach ($routing_key as $rk) {
                $queue->bind($exchange_name, $rk);
            }
        } else {
            $queue->bind($exchange_name, $routing_key);
        }

        return $queue;
    }

    public function dirct_push($exchange_name, $queue, $routing_key, $msg, $ttl = null)
    {

        $ex = $this->get_exchage($exchange_name);
        $this->get_queue($queue, $exchange_name, $routing_key);
        $message_attrs = [];

        if ($this->is_delay_message) {
            $message_attrs = [
                'headers' => ['x-delay' => intval(1000 * ($ttl ?? 1))],
            ];
        }

        $ex->publish($msg, $routing_key ?? 'error', AMQP_NOPARAM, $message_attrs);
    }

    public function push_mq($message)
    {
        $_exchange = $this->get_exchage();
        $_exchange->publish($message, '');
    }

    public function set_is_delay_message($val)
    {
        $this->is_delay_message = $val;
    }

    public function consume($queue_name, $ex_name, $routing_key, $call_back)
    {
        $ex = $this->get_exchage($ex_name);
        $que = $this->get_queue($queue_name, $ex_name, $routing_key);
        $que->consume($call_back, AMQP_NOPARAM);
        $this->_channel->close();
        $this->_connection->disconnect();
    }

    public function mq_get($queue_name, $ex_name, $routing_key)
    {

        $ex = $this->get_exchage($ex_name);
        $que = $this->get_queue($queue_name, $ex_name, $routing_key);

        
        $k = $que->get();
        if (!$k) {
            writeln('no message');
            return;
        }

        $msg = sprintf("from %s exchange,%s routing key,body is :%s", $k->getExchangeName(), $k->getRoutingKey(), $k->getBody());
        writeln($msg);
        $que->nack($k->getDeliveryTag(),AMQP_REQUEUE); //nack后，如果有设置死信队列，则跑到死信队列，否则会丢失？
    }
}
